package net.bw.realtime.jtp.ods.job;

import net.bw.realtime.jtp.common.utils.KafkaUtil;
import net.bw.realtime.jtp.common.utils.MysqlCdcUtil;
import net.bw.realtime.jtp.ods.function.CDCDataEtlMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-02
 */
public class JtpCDCMallDataJob {

    public static void main(String[] args) throws Exception {

        // 1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //  设置并行度
        env.setParallelism(1);
        //  开启检查点
        env.enableCheckpointing(5000L);

        // 2.数据源
        DataStream<String> mysqlSource = cdcMysqlSource(env);
        // 输出
        //mysqlSource.print();

        // 3.数据清洗
         DataStream<String> cleanDataStream = mysqlSource.map(new CDCDataEtlMapFunction());
        //输出
        cleanDataStream.print();

        // 4.数据写入Kafka
        KafkaUtil.producerKafka(cleanDataStream, "topic-mall");

        // 启动程序
        env.execute("JtpCDCMallDataJob");
    }

    //  通过FlinkCDC采集MySQL数据库的数据
    //  维度表数据先进行全量采集，在进行增量采集
    //  事实表数据进行增量采集
    private static DataStream<String> cdcMysqlSource(StreamExecutionEnvironment env) {

        // 1.维度表数据
        String[] dim_list = new String[]{
                "base_category1", "base_category2", "base_category3", "base_dic", "base_province", "base_region",
                "base_trademark", "user_info", "sku_info", "spu_info", "activity_info", "activity_rule", "activity_sku"
        };

        // 传入参数获取维度表数据
        DataStream<String> dimStream = MysqlCdcUtil.cdcMysqlInitial(env, "jtp_mall", dim_list);
        // 输出
        //dimStream.print();

        // 2.事实表数据
        String[] dwd_list = new String[]{
                "cart_info", "comment_info", "coupon_info", "coupon_range", "coupon_use", "favor_info", "order_detail",
                "order_detail_activity", "order_detail_coupon", "order_info", "order_refund_info", "order_status_log",
                "payment_info", "refund_payment"
        };

        //  获取事实表数据
        DataStream<String> dwdStream = MysqlCdcUtil.cdcMysqlEarliest(env, "jtp_mall", dwd_list);
        // 输出
        //dwdStream.print();

        // 合并数据
        DataStream<String> unionStream = dimStream.union(dwdStream);
        // 输出
        //unionStream.print();

        //  返回数据
        return unionStream;
    }

}
